package com.shujia.flink.window

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

object Demo4WindowProcess {

  def main(args: Array[String]): Unit = {

    /*

001,a
001,b
001,c
001,d

     */


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")


    val consumer = new FlinkKafkaConsumer[String]("test_topic1", new SimpleStringSchema(), properties)

    consumer.setStartFromLatest()

    val kafkaDS: DataStream[String] = env.addSource(consumer)

    val kvDS: KeyedStream[(String, String), String] = kafkaDS
      .map(line => {
        val split: Array[String] = line.split(",")
        (split(0), split(1))
      })
      .keyBy(_._1)


    //如果5秒没有数据过来进行计算
    val windowDS: WindowedStream[(String, String), String, TimeWindow] = kvDS
      .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))


    //使用flink 底层api 操作底层事件，事件和状态
    val countDS: DataStream[(String, String)] = windowDS.process(new MyWindowProcessFunction)


    countDS.print()


    env.execute()


  }

}

class MyWindowProcessFunction extends ProcessWindowFunction[(String, String), (String, String), String, TimeWindow] {
  /**
    * 处理窗口内数据的方法, 每一个key  每一个窗口处理一次
    *
    * @param key      数据的key
    * @param context  上下文对象
    * @param elements 一个key 在一个窗口内的所有数据
    * @param out      用于将数据发送到下游
    */
  override def process(key: String, context: Context, elements: Iterable[(String, String)], out: Collector[(String, String)]): Unit = {

    //如果一个人操作顺序是a,b,c,d ,发出警告
    val flag = List("a", "b", "c", "d")

    val list: List[String] = elements.toList.map(_._2)

    var tage = true

    if (list.length >= 4) {
      var i = 0
      while (i < flag.length) {

        val f = flag(i)
        val l = list(i)

        if (f != l) {
          tage = false
        }

        i += 1
      }
    }

    if (tage) {
      //发出警告
      out.collect((key, list.mkString("|")))
    }

  }
}

